package org.lib.kafkahello;

import java.util.Map;
import java.util.function.Consumer;

import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import lombok.extern.slf4j.Slf4j;

/**
 * 检查Kafka部分Bean的信息
 * @author ben
 * @date 2021-08-29 16:15:13 CST
 */
@RestController
@RequestMapping(value="/check")
@Slf4j
public class KafkaCheck {

	@Autowired
	private KafkaTemplate<String, Object> kafkaTemplate;
	
	@Autowired
	private KafkaAdmin kafkaAdmin;
	
	@Resource(name="kafkaProducerFactory")
	private DefaultKafkaProducerFactory kafkaProducerFactory;
	
	@Resource(name="kafkaConsumerFactory")
	private DefaultKafkaConsumerFactory kafkaConsumerFactory;
	
	private static final Consumer<Object> cs = System.out::println;
	
	@GetMapping(value="/kafkaTemplate")
	public Boolean kafkaTemplate() {
		cs.accept("check kafkaTemplate:");
		cs.accept(kafkaTemplate);
		cs.accept(kafkaTemplate.getDefaultTopic());
		cs.accept(kafkaTemplate.getTransactionIdPrefix());
		cs.accept(kafkaTemplate.getMessageConverter());
		cs.accept(kafkaTemplate.getProducerFactory());
		return true;
	}

	@GetMapping(value="/kafkaAdmin")
	public Boolean kafkaAdmin() {
		cs.accept("check kafkaAdmin:");
		cs.accept(kafkaAdmin);
		cs.accept("getConfigurationProperties:");
		Map<String,Object> config = kafkaAdmin.getConfigurationProperties();
		config.forEach((k, v)->{
			cs.accept("k=" + k + ", v=" + v);
		});
		return true;
	}

	@GetMapping(value="/kafkaProducerFactory")
	public Boolean kafkaProducerFactory() {
		cs.accept("check kafkaProducerFactory:");
		cs.accept(kafkaProducerFactory);
		cs.accept(kafkaProducerFactory.getConfigurationProperties());
		cs.accept(kafkaProducerFactory.getKeySerializerSupplier());
		cs.accept(kafkaProducerFactory.getValueSerializerSupplier());
		return true;
	}

	@GetMapping(value="/kafkaConsumerFactory")
	public Boolean kafkaConsumerFactory() {
		cs.accept("check kafkaConsumerFactory:");
		cs.accept(kafkaConsumerFactory);
		cs.accept(kafkaConsumerFactory.getConfigurationProperties());
		cs.accept(kafkaConsumerFactory.getKeyDeserializer());
		cs.accept(kafkaConsumerFactory.getValueDeserializer());
		return true;
	}
	
}
